package com.reload.core;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

/**
 * @Date: 2024/4/27 0027 - 12:15
 * @Version: 17
 * @Author: Reloading
 * @Description: 响应式核心 Subscribe
 */
public class SubscribeCore {
    public static void main(String[] args) {
        /*
         关于 doOnXXX()和OnErrorXXX()
         doOnXXX() 发生这个事件的时候产生一个回调 通知你(不能改变)
         OnErrorXXX() 发生这个时间后执行的一个动作，可以改变元素、信号
         */
        Flux<String> flux = Flux.range(1, 7)
                .map(i -> {
                    // 模拟异常
                    // if (i == 6) i = 10 / 0;
                    return "哈哈：" + i;
                })
                // 流错误的时候把错误吃掉转为正常信号
                .onErrorComplete();

        flux.subscribe(System.out::println, throwable -> {
            System.out.println(throwable.getMessage());
        }, () -> System.out.println("流完成"));

        // 自定义消费者
        // 生命周期钩子 流的生命周期的钩子可以传递给订阅者
        flux.subscribe(new BaseSubscriber<String>() {
            // 流被订阅的时候触发
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("自定义消费者 -- 绑定了...." + subscription);
                // 找发布者要数据
                // 请求N次数据
                request(1);
                // 要无限的数据
                // requestUnbounded();
            }

            // 数据到达
            @Override
            protected void hookOnNext(String value) {
                System.out.println("自定义消费者 -- 数据到达：" + value);
                if (value.equals("哈哈：3")) {
                    // 取消流
                    cancel();
                }
                request(1);
            }

            // 流完成
            @Override
            protected void hookOnComplete() {
                System.out.println("自定义消费者 -- 流完成");
            }

            // 异常执行
            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("自定义消费者 -- 流异常");
            }

            // 取消执行
            @Override
            protected void hookOnCancel() {
                System.out.println("自定义消费者 -- 流完成");
            }

            // 最终执行
            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("自定义消费者 -- 流最终执行");
            }
        });
    }
}
